import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;


public class MR {
	
	private static String TABELA_MSG = "mensagem";
	private static String TABELA_USER = "usuario";
	private static byte [] ATRIBUTOS = Bytes.toBytes("atributos");
	private static byte [] TEXTO = Bytes.toBytes("text");

	public static class Map extends
			TableMapper<ImmutableBytesWritable, ImmutableBytesWritable> {

	    protected void map(ImmutableBytesWritable key, Result result, Context context) 
	    	throws IOException, InterruptedException {
	    	String userId = Bytes.toString(key.get()).substring(0, 36);
	    	
	    	HTable htable = new HTable(TABELA_USER);
			Get get = new Get(Bytes.toBytes(userId));
			if (htable.get(get).isEmpty()) {
				ImmutableBytesWritable userIdNotFound = new ImmutableBytesWritable(Bytes.toBytes(userId));
				context.write(userIdNotFound, userIdNotFound);
			}
		}
	}

	public static class Reduce
			extends TableReducer<ImmutableBytesWritable, ImmutableBytesWritable, NullWritable> {

		protected void reduce(ImmutableBytesWritable userId,
				Iterable<ImmutableBytesWritable> userIds, Context context)
				throws IOException, InterruptedException {
			int count = 0;
			for (ImmutableBytesWritable id : userIds) {
				count++;
			}
			System.out.println("===>" + Bytes.toString(userId.get()) + " tinha " + count + " mensagens!");
		}
	}

	public MR() {

	}

	public Job createSubmittableJob(Configuration conf) throws IOException {
		Job job = new Job(conf, "job");
		job.setJarByClass(this.getClass());

		String tableName = new String(TABELA_MSG);

		Scan scan = new Scan();
		scan.addColumn(ATRIBUTOS, TEXTO);

		job.setNumReduceTasks(1);

		TableMapReduceUtil.initTableMapperJob(tableName, scan, Map.class,
						ImmutableBytesWritable.class,
						ImmutableBytesWritable.class, job);
		TableMapReduceUtil.initTableReducerJob(tableName, Reduce.class, job);

		return job;
	}

	public static void main(String[] args) throws IOException,
			InterruptedException, ClassNotFoundException {
		Configuration conf = HBaseConfiguration.create();
		MR mr = new MR();
		mr.createSubmittableJob(conf).waitForCompletion(true);
	}

}
